package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/4/25 10:02
 */
public class Dwd_04_DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_DwdTradeOrderDetail().init(
            30004,
            2,
            "Dwd_04_DwdTradeOrderDetail"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv, "Dwd_04_DwdTradeOrderDetail");
        // 2. 过滤出详情数据
        Table orderDetail = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['sku_name'] sku_name, " +
                "data['create_time'] create_time, " +
                "data['sku_num'] sku_num, " +
                "cast(cast(data['sku_num'] as decimal(16,2)) * " +
                "   cast(data['order_price'] as decimal(16,2)) as String) split_original_amount, " +
                "data['split_total_amount'] split_total_amount, " +  // 减免后
                "data['split_activity_amount'] split_activity_amount, " +
                "data['split_coupon_amount'] split_coupon_amount, " +
                "ts  " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);
        
        //orderDetail.execute().print();  // 这个是阻塞式方法: 表中的数据没有输出完, 则不会向下执行
        
        // 3. 过滤出订单数据
        Table orderInfo = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['province_id'] province_id " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='order_info' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        
        //orderInfo.execute().print();
        // 4. 详情活动
        Table orderDetailActivity = tEnv.sqlQuery(
            "select " +
                "data['order_detail_id'] order_detail_id,  " +
                "data['activity_id'] activity_id,  " +
                "data['activity_rule_id'] activity_rule_id  " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='order_detail_activity' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail_activity", orderDetailActivity);
        
        // 5. 详情优惠券
        Table orderDetailCoupon = tEnv.sqlQuery(
            "select " +
                "data['order_detail_id'] order_detail_id,  " +
                "data['coupon_id'] coupon_id  " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='order_detail_coupon' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);
        
        // 6. 4 张表 join
        Table result = tEnv.sqlQuery(
            "select " +
                "od.id,  " +
                "od.order_id,  " +
                "oi.user_id,  " +
                "od.sku_id,  " +
                "od.sku_name,  " +
                "oi.province_id,  " +
                "act.activity_id,  " +
                "act.activity_rule_id,  " +
                "cou.coupon_id,  " +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id,  " +
                "od.create_time,  " +
                "od.sku_num,  " +
                "od.split_original_amount,  " +
                "od.split_activity_amount,  " +
                "od.split_coupon_amount,  " +
                "od.split_total_amount,  " +
                "od.ts   " +
                "from order_detail od " +
                "join order_info oi on od.order_id=oi.id " +
                "left join order_detail_activity act on od.id=act.order_detail_id " +
                "left join order_detail_coupon cou on od.id=cou.order_detail_id");
        
        
        // 7. 写出到 dwd 层
        tEnv.executeSql(
            "create table dwd_trade_order_detail(  " +
                "id string,  " +
                "order_id string,  " +
                "user_id string,  " +
                "sku_id string,  " +
                "sku_name string,  " +
                "province_id string,  " +
                "activity_id string,  " +
                "activity_rule_id string,  " +
                "coupon_id string,  " +
                "date_id string,  " +
                "create_time string,  " +
                "sku_num string,  " +
                "split_original_amount string,  " +
                "split_activity_amount string,  " +
                "split_coupon_amount string,  " +
                "split_total_amount string,  " +
                "ts bigint, " +
                "primary key(id)not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        
        result.executeInsert("dwd_trade_order_detail");
    }
    
}
/*
交易域下单事务事实表
    粒度: 最小粒度, 商品
    
    order_detail   insert
       inner join
    order_info     insert
       left join
    order_detail_activity  insert
       left join
    order_detail_coupon   insert
    
    join的 ttl 问题:
        5s
        
----
在 sql 中, 需要同时打印两张表的数据:
 1. 把表转成流打印(流的打印不会阻塞)
 2. 使用 print connector
 */